package com.zl.ots;

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.core.utils.Base64;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.model.search.SearchQuery;
import com.alicloud.openservices.tablestore.model.search.SearchRequest;
import com.alicloud.openservices.tablestore.model.search.SearchResponse;
import com.alicloud.openservices.tablestore.model.search.groupby.GroupBy;
import com.alicloud.openservices.tablestore.model.search.groupby.GroupByBuilders;
import com.alicloud.openservices.tablestore.model.search.groupby.GroupByField;
import com.alicloud.openservices.tablestore.model.search.groupby.GroupByFieldResultItem;
import com.zl.ots.page.OtsPage;
import com.zl.ots.page.OtsQuery;
import com.zl.ots.utils.DateUtils;
import com.zl.ots.utils.JsonUtil;
import com.zl.ots.utils.LocalDateUtils;
import com.zl.ots.utils.NamingConverterUtil;
import com.zl.ots.wrapper.OtsQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;

import static com.zl.ots.page.OtsPage.FIRST_OTS_TOKEN;
import static com.zl.ots.utils.DateUtils.DATE_TIME_PATTERN;


/**
 * 表格存储相关工具
 */
@Slf4j
@Component
public class OtsClient {

    protected static SyncClient client;
    protected static OtsRedisUtil redisUtil;

    @Resource
    public void setClient(SyncClient client){
        OtsClient.client = client;
    }

    @Resource
    public void setRedisUtil(OtsRedisUtil redisUtil){
        OtsClient.redisUtil = redisUtil;
    }

    /**
     * 同步ots
     * @param data      同步的对象
     * @param tableName 表名
     * @param idField   主键字段名
     * @param id        主键值
     */
    public static void syncOts(Object data,String tableName,String idField,String id){
        //构造主键。
        PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
        primaryKeyBuilder.addPrimaryKeyColumn(idField, PrimaryKeyValue.fromString(id));
        PrimaryKey primaryKey = primaryKeyBuilder.build();
        //设置数据表名称
        RowUpdateChange rowUpdateChange = new RowUpdateChange(tableName, primaryKey);
        //同步
        putFieldValue(rowUpdateChange,data,idField);
        //此操作表格存储会判断，如果id存在则跟新 不存在则新增
        client.updateRow(new UpdateRowRequest(rowUpdateChange));
    }

    /**
     * 同步属性到表格存储
     * @param rowUpdateChange 更新ots
     * @param obj   对象
     * @param pkField 逐渐属性
     */
    public static void putFieldValue(RowUpdateChange rowUpdateChange,Object obj,String pkField){
        Field[] declaredFields = obj.getClass().getDeclaredFields();
        for(Field field : declaredFields){
            if(field.getName().equals(pkField)){
                //如果是主键跳过
                continue;
            }
            Object o = null;
            try {
                field.setAccessible(true);
                o = field.get(obj);
            } catch (IllegalAccessException e) {
                continue;
            }
            if(o == null){
                continue;
            }
            ColumnValue value = getColumnValue(o);
            String fieldName = NamingConverterUtil.camelToSnake(field.getName());
            rowUpdateChange.put(new Column(fieldName, value));
        }
    }

    public static void setPrimaryKeyFieldObj(Field field,Object o,PrimaryKeyValue keyValue){
        Class<?> type = field.getType();
        Object fieldValue = null;
        if(type.isAssignableFrom(String.class)){
            fieldValue = keyValue.asString();
        }else if(type.isAssignableFrom(Long.class)){
            fieldValue = keyValue.asLong();
        }else{
            throw new RuntimeException("主键类型错误:"+field.getName());
        }
        try {
            field.setAccessible(true);
            field.set(o,fieldValue);
        } catch (Exception e) {
            log.error("表格存储查询设置主键属性值失败！{}" , field.getName(),e);
            throw new RuntimeException(e);
        }
    }

    public static void setFieldObj(Field field,Object o,ColumnValue value){
        Class<?> type = field.getType();
        Object fieldValue = null;
        if(type.isAssignableFrom(String.class)){
            fieldValue = value.asString();
        }else if(type.isAssignableFrom(Boolean.class)){
            fieldValue = value.asBoolean();
        }else if(type.isAssignableFrom(Date.class)){
            fieldValue = DateUtils.parseDate(value.asString(),DATE_TIME_PATTERN);
        }else if(type.isAssignableFrom(Long.class)){
            fieldValue = value.asLong();
        }else if(type.isAssignableFrom(BigDecimal.class)){
            fieldValue = BigDecimal.valueOf(value.asDouble());
        }else if(type.isAssignableFrom(Integer.class)){
            fieldValue = Long.valueOf(value.asLong()).intValue();
        }else if(type.isAssignableFrom(Double.class)){
            fieldValue = value.asDouble();
        }else if(type.isAssignableFrom(Float.class)){
            fieldValue = Double.valueOf(value.asDouble()).floatValue();
        }else if(type.isAssignableFrom(Short.class)){
            fieldValue = Long.valueOf(value.asLong()).shortValue();
        }else if(type.isAssignableFrom(Byte.class)){
            fieldValue = Long.valueOf(value.asLong()).byteValue();
        }else if(type.isAssignableFrom(LocalDateTime.class)){
            fieldValue = LocalDateUtils.parseDateTime(value.asString(),DATE_TIME_PATTERN);
        }else if(type.isAssignableFrom(LocalDate.class)){
            fieldValue = LocalDateUtils.parseDate(value.asString());
        } else{
            fieldValue = JsonUtil.toBean(value.asString(),type);
        }
        try {
            field.setAccessible(true);
            field.set(o,fieldValue);
        } catch (Exception e) {
            log.error("表格存储查询设置属性值失败！{}" , field.getName(),e);
            throw new RuntimeException(e);
        }
    }

    public static ColumnValue getColumnValue(Object o){
        ColumnValue value = null;
        if(o instanceof Long || o instanceof Integer || o instanceof Short || o instanceof Byte
                || o instanceof BigInteger){
            value = ColumnValue.fromLong(((Number) o).longValue());
        }else if(o instanceof BigDecimal || o instanceof Float || o instanceof Double){
            value = ColumnValue.fromDouble(((Number) o).doubleValue());
        } else if(o instanceof CharSequence){
            value = ColumnValue.fromString(String.valueOf(o));
        }else if(o instanceof Boolean){
            value = ColumnValue.fromBoolean((Boolean) o);
        }else if(o instanceof Date){
            value = ColumnValue.fromString(DateUtils.format((Date)o,DATE_TIME_PATTERN));
        }else {
            value = ColumnValue.fromString(JsonUtil.toJsonStr(o));
        }
        return value;
    }



    private static <T>List<T> buildTFromRows(List<Row> rows,Class<T> clazz,String idFieldName){
        if(ObjectUtils.isEmpty(rows)){
            return new ArrayList<>();
        }
        List<T> list = new ArrayList<>();
        //获取表属性列名
        Field[] otsFields = clazz.getDeclaredFields();
        for(Row row : rows){
            //主键信息
            String otsIdField = NamingConverterUtil.camelToSnake(idFieldName);
            PrimaryKeyColumn primaryKeyColumn = row.getPrimaryKey().getPrimaryKeyColumn(otsIdField);
            if(ObjectUtils.isEmpty(primaryKeyColumn)){
                log.info("表格存储主键属性不存在:{}",idFieldName);
                continue;
            }
            PrimaryKeyValue idValue = primaryKeyColumn.getValue();
            try {
                T t = clazz.newInstance();
                for(Field otsField : otsFields){
                    String otsFieldName = NamingConverterUtil.camelToSnake(otsField.getName());
                    Column column = row.getLatestColumn(otsFieldName);
                    if(column != null){
                        ColumnValue value = column.getValue();
                        setFieldObj(otsField,t,value);
                        continue;
                    }
                    //如果字段不存在，查询是否是主键字段
                    String name = otsField.getName();
                    if(idFieldName.equals(name)){
                        //主键字段
                        setPrimaryKeyFieldObj(otsField,t,idValue);
                    }
                }
                list.add(t);
            } catch (Exception e) {
                log.error("表格存储实例化对象失败！{}",clazz.getName());
                throw new RuntimeException(e);
            }
        }
        return list;
    }

    /**
     * 统计行数
     * @param wrapper   查询对象wrapper
     * @param tableName  表名
     * @param indexName  索引名
     * @return  行数
     * @param <T> 查询对象
     */
    public static <T>Long queryTotal(OtsQueryWrapper<T> wrapper, String tableName, String indexName){
        SearchQuery searchQuery = wrapper.build();
        //统计行数，limit设置0
        searchQuery.setLimit(0);
        //设置返回总数
        searchQuery.setGetTotalCount(Boolean.TRUE);
        //组装请求
        SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
        //开始查询 返回
        SearchResponse resp = client.search(searchRequest);
        return resp.getTotalCount();
    }

    /**
     * 分组聚合统计
     */
    public static <T>Map<String,Long> queryTotalGroup(OtsQueryWrapper<T> wrapper,String tableName,String indexName,String groupField){
        SearchQuery searchQuery = wrapper.build();
        //统计行数，limit设置0
        searchQuery.setLimit(0);
        //分组
        List<GroupBy> groupByList = new ArrayList<>();
        GroupByField groupBy = GroupByBuilders.groupByField("group_by", groupField).size(50).build();
        groupByList.add(groupBy);
        searchQuery.setGroupByList(groupByList);
        //组装请求
        SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
        //开始查询 返回
        SearchResponse resp = client.search(searchRequest);
        List<GroupByFieldResultItem> resultItems = resp.getGroupByResults().getAsGroupByFieldResult("group_by")
                .getGroupByFieldResultItems();
        if(!ObjectUtils.isEmpty(resultItems)){
            Map<String,Long> map = new HashMap<>();
            resultItems.forEach(r->{
                map.put(r.getKey(),r.getRowCount());
            });
            return map;
        }
        return new HashMap<>();
    }

    /**
     * 分页查询表格存储
     * @param wrapper   查询构造
     * @param tableName  表名
     * @param indexName  多元索引名
     * @param idFieldName  主键字段名
     * @param query 分页参数
     * @param querySign 查询签名，分页查询使用，点击下页码，需要传递此参数，一个分页周期签名需保持一致，否则分页失败
     * @return 结果集
     * @param <T> 返回直接对象
     */
    public static <T>List<T> queryPage(OtsQueryWrapper<T> wrapper, String tableName, String indexName, String idFieldName, OtsQuery query, Class<T> clazz,String querySign){
        SearchQuery searchQuery = wrapper.build();
        searchQuery.setLimit(query.getPageSize());
        //分页条件
        OtsPage page = new OtsPage(redisUtil,querySign,query.getNowToken());
        String nowToken = page.getNowToken();
        if(!ObjectUtils.isEmpty(nowToken) && !FIRST_OTS_TOKEN.equals(nowToken)){
            searchQuery.setToken(Base64.fromBase64String(nowToken));
        }
        String beforeToken = page.getBeforeToken();
        query.setBeforeToken(beforeToken);
        //设置返回总数s
        searchQuery.setGetTotalCount(Boolean.TRUE);
        //组装请求
        SearchRequest searchRequest = new SearchRequest(tableName, indexName, searchQuery);
        SearchRequest.ColumnsToGet columnsToGet = new SearchRequest.ColumnsToGet();
        //设置为返回所有列。
        columnsToGet.setReturnAllFromIndex(true);
        searchRequest.setColumnsToGet(columnsToGet);
        //开始查询
        SearchResponse resp = client.search(searchRequest);
        List<Row> rows = resp.getRows();
        long totalCount = resp.getTotalCount();
        query.setTotalCount((int)totalCount);
        //下一个token 如果下一个token为空，则代表最后一页
        byte[] nextToken = resp.getNextToken();
        if(nextToken != null){
            String base64Token = Base64.toBase64String(nextToken);
            query.setNextToken(base64Token);
            page.addNextToken(base64Token);
        }else{
            query.setNextToken(null);
        }
        return buildTFromRows(rows,clazz,idFieldName);
    }

}
